/*
 * Copyright 2019-present HiveMQ GmbH
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hivemq.extensions.rbac.redis;

import com.hivemq.extension.sdk.api.auth.SimpleAuthenticator;
import com.hivemq.extension.sdk.api.auth.parameter.SimpleAuthInput;
import com.hivemq.extension.sdk.api.auth.parameter.SimpleAuthOutput;
import com.hivemq.extension.sdk.api.auth.parameter.TopicPermission;
import com.hivemq.extension.sdk.api.client.parameter.Listener;
import com.hivemq.extension.sdk.api.packets.auth.DefaultAuthorizationBehaviour;
import com.hivemq.extension.sdk.api.packets.connect.ConnackReasonCode;
import com.hivemq.extensions.rbac.redis.configuration.entities.ExtensionRedisConfig;
import com.hivemq.extensions.rbac.redis.utils.RedisKeyOnecCredentialsValidator;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Optional;

/**
 * 2. 身份验证器
 */
class RedisAuthAuthenticator implements SimpleAuthenticator {

    private final @NotNull RedisKeyOnecCredentialsValidator credentialsValidator;
    private final @NotNull ExtensionRedisConfig extensionRedisConfig;
    private static final @NotNull Logger LOG = LoggerFactory.getLogger(RedisAuthAuthenticator.class);

    RedisAuthAuthenticator(
            final @NotNull RedisKeyOnecCredentialsValidator credentialsValidator, final @NotNull ExtensionRedisConfig extensionRedisConfig) {
        this.credentialsValidator = credentialsValidator;
        this.extensionRedisConfig = extensionRedisConfig;
    }

    @Override
    public void onConnect(
            final @NotNull SimpleAuthInput simpleAuthInput, final @NotNull SimpleAuthOutput simpleAuthOutput) {
        final boolean nextExtensionInsteadOfFail = extensionRedisConfig.isNextExtensionInsteadOfFail();
        final Optional<Listener> connectedListenerOptional = simpleAuthInput.getConnectionInformation().getListener();


        final Optional<String> userNameOptional = simpleAuthInput.getConnectPacket().getUserName();
        final Optional<ByteBuffer> passwordOptional = simpleAuthInput.getConnectPacket().getPassword();
        final String clientId = simpleAuthInput.getClientInformation().getClientId();

        // 检查用户名
        if (userNameOptional.isEmpty() || passwordOptional.isEmpty()) {
            // 如果允许匿名用户
            if (nextExtensionInsteadOfFail) {
                simpleAuthOutput.nextExtensionOrDefault();
                return;
            }
            simpleAuthOutput.failAuthentication(ConnackReasonCode.BAD_USER_NAME_OR_PASSWORD,
                    "Authentication failed because username or password are missing");
            return;
        }
        final String userName = userNameOptional.get();

        // 检查客户端id - 通配符检查
        if (clientId.contains("#") || clientId.contains("+")) {
            // 未通过检查
            if (nextExtensionInsteadOfFail) {
                simpleAuthOutput.nextExtensionOrDefault();
                return;
            }
            // 不允许非法的ID
            simpleAuthOutput.failAuthentication(ConnackReasonCode.CLIENT_IDENTIFIER_NOT_VALID,
                    "The characters '#' and '+' are not allowed in the client identifier");
            return;
        }

        // 检查用户名 - 通配符检查
        if (userName.contains("#") || userName.contains("+")) {
            // 未通过检查
            if (nextExtensionInsteadOfFail) {
                simpleAuthOutput.nextExtensionOrDefault();
                return;
            }
            // 不允许通配符的用户名
            simpleAuthOutput.failAuthentication(ConnackReasonCode.BAD_USER_NAME_OR_PASSWORD,
                    "The characters '#' and '+' are not allowed in the username");
            return;
        }

        final String connectedListenerName = connectedListenerOptional.get().getName();
        String password = convertOptionalByteBufferToString(passwordOptional);
        LOG.debug("[RedisAuth-{}]开始身份验证->uame:{},clientId:{},password:{}", connectedListenerName, userName, clientId, password);
        // 检查我们是否有用户名/密码组合的角色
        String key = userName + "/" + clientId + "/" + password;

        if(!userName.contains("admin") && clientId.startsWith("qr")){
            if(!credentialsValidator.exists(key)){
                LOG.info("拒绝连接:{}", key);
                simpleAuthOutput.failAuthentication(ConnackReasonCode.NOT_AUTHORIZED,
                        "Authentication failed because of invalid credentials");
            }
            else{
                LOG.info("通过连接:{}", key);
                simpleAuthOutput.getDefaultPermissions().add(credentialsValidator.getPermission(key, TopicPermission.Qos.TWO, TopicPermission.MqttActivity.SUBSCRIBE));
                simpleAuthOutput.getDefaultPermissions().add(credentialsValidator.getPermission("WillMsg", TopicPermission.Qos.ZERO, TopicPermission.MqttActivity.PUBLISH));
                simpleAuthOutput.getDefaultPermissions().setDefaultBehaviour(DefaultAuthorizationBehaviour.DENY);
                simpleAuthOutput.authenticateSuccessfully();
            }
        }
        else{
            // 进入下一步验证链
            simpleAuthOutput.nextExtensionOrDefault();
        }
    }

    /**
     * Optional<ByteBuffer> 转 string
     * @param optionalByteBuffer Optional<ByteBuffer>
     * @return 目标字符串
     */
    public static String convertOptionalByteBufferToString(Optional<ByteBuffer> optionalByteBuffer) {
        if (optionalByteBuffer.isPresent()) {
            ByteBuffer byteBuffer = optionalByteBuffer.get();
            ByteBuffer writableBuffer = ByteBuffer.allocate(byteBuffer.remaining());
            writableBuffer.put(byteBuffer);
            writableBuffer.flip();
            byte[] array = writableBuffer.array();
            return new String(array, StandardCharsets.UTF_8);
        } else {
            return "";
        }
    }
}
